package spring.cloud.tasks.tasks_assign_manager;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Setter
@Getter
public class TasksManagerConnectionStateListener implements org.apache.curator.framework.state.ConnectionStateListener {


    private String threadName;
    private TasksManager taskManager;
    private ExecutorService executorService;
    //
    private AtomicBoolean shutdown = new AtomicBoolean(false);
    //
    private AtomicBoolean connected = new AtomicBoolean(false);
    private AtomicBoolean stopped = new AtomicBoolean(false);

    public TasksManagerConnectionStateListener(TasksManager taskManager, final String threadName) {
        this.taskManager = taskManager;
        this.threadName = threadName;
        this.executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, threadName);
                thread.setDaemon(false);
                thread.setPriority(Thread.NORM_PRIORITY);
                return thread;
            }
        });
    }

    public void restart() {

    }


    public void stop() {
        try {

            taskManager.taskItemAssignService.shutdown();
        } catch (InterruptedException e) {
            log.info("stop interrupted");
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("stop error", e);
        }
    }

    //
    @Override
    public void stateChanged(final CuratorFramework curatorFramework, final ConnectionState connectionState) {
        // 使用single thread executor严格保证ZK事件执行的顺序性，避免并发性问题
        if (ConnectionState.RECONNECTED == connectionState) {
            connected.set(true);
            executorService.submit(new Runnable() {//
                @Override
                public void run() {
                    if (stopped.compareAndSet(true, false)) {
                        log.info("try to restart for zk reconnected");
                        restart();
                    }
                }
            });
        } else if (ConnectionState.SUSPENDED == connectionState) {
            connected.set(false);
            final long sessionId = getSessionId(curatorFramework);
            executorService.submit(new Runnable() {//
                @Override
                public void run() {
                    do {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                        if (shutdown.get()) {
                            return;
                        }
                        long newSessionId = getSessionId(curatorFramework);
                        if (sessionId != newSessionId) {
                            log.info("try to stop for zk lost");
                            stop();
                            stopped.set(true);
                            return;
                        }
                    } while (!shutdown.get() && !connected.get());
                }
            });
        } else {
            //...
        }
    }

    private long getSessionId(CuratorFramework curatorFramework) {
        long sessionId;
        try {
            sessionId = curatorFramework.getZookeeperClient().getZooKeeper().getSessionId();
        } catch (Exception e) {// NOSONAR
            return -1;
        }
        return sessionId;
    }

    public void shutdownNowUntilTerminated() throws InterruptedException {
        shutdown.set(true);
        while (true) {
            executorService.shutdownNow();
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                if (!executorService.isTerminated()) {
                    log.error("shutdownNowUntilTerminated is interrupted, but the {} is not terminated", threadName);
                }
                throw e;
            }
            if (executorService.isTerminated()) {
                return;
            }
        }
    }

}
